package com.example.testkafka.consumer.controller;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.*;


@RestController
@RequestMapping("/consumer")
public class ConsumerController {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerController.class);

    @Autowired
    private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    @RequestMapping("/testStartConsumer")
    @ResponseBody
    public Map<String,Object> testStartConsumer() {

        Map<String,Object> appResult = new HashMap<>();
        kafkaListenerEndpointRegistry.start();
        appResult.put("status","SUCCESS");
        return appResult;
    }

    @RequestMapping("/testStopConsumer")
    @ResponseBody
    public Map<String,Object> testProducer() {

        Map<String,Object> appResult = new HashMap<>();
        kafkaListenerEndpointRegistry.stop();
        appResult.put("status","SUCCESS");
        return appResult;
    }

    @RequestMapping("/testPause")
    @ResponseBody
    public Map<String,Object> testPause(String id) {

        Map<String,Object> appResult = new HashMap<>();
        MessageListenerContainer c1 = kafkaListenerEndpointRegistry.getListenerContainer(id);
        Collection<MessageListenerContainer> c2 = kafkaListenerEndpointRegistry.getAllListenerContainers();
        Collection<MessageListenerContainer> c3 = kafkaListenerEndpointRegistry.getListenerContainers();
        Set<String> ids = kafkaListenerEndpointRegistry.getListenerContainerIds();
        c1.pause();
        appResult.put("status","SUCCESS");
        return appResult;
    }

    @RequestMapping("/testResume")
    @ResponseBody
    public Map<String,Object> testResume(String id) {

        Map<String,Object> appResult = new HashMap<>();
        MessageListenerContainer c1 = kafkaListenerEndpointRegistry.getListenerContainer(id);
        c1.resume();
        appResult.put("status","SUCCESS");
        return appResult;
    }

//    @RequestMapping("/testSeek")
//    @ResponseBody
//    public Map<String,Object> testSeek() {
//
//        Map<String,Object> appResult = new HashMap<>();
//        MessageListenerContainer c1 = kafkaListenerEndpointRegistry.getListenerContainer("helloworld");
//        ConcurrentMessageListenerContainer c = (ConcurrentMessageListenerContainer) c1;
//        List<KafkaMessageListenerContainer> cs = c.getContainers();
//        for(KafkaMessageListenerContainer o : cs){
//            o.
//        }
//        appResult.put("status","SUCCESS");
//        return appResult;
//    }
}
